package com.ljl.question3;

import jakarta.annotation.Resource;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.config.EnableIntegration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.handler.LoggingHandler;
import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
import org.springframework.integration.mqtt.support.DefaultPahoMessageConverter;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.MessageHandler;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.StringUtils;

import java.util.Locale;
import java.util.UUID;

/**
 * mqtt业务类
 *
 * @author ljl
 */
@Configuration
@EnableIntegration
public class MqttServiceImpl implements MqttService {
    @Resource
    private MqttPahoClientFactory clientFactory;

    @Resource
    private MqttProperties properties;

    /**
     * 消息发布
     *
     * @return /
     */
    @Bean
    public IntegrationFlow mqttOutFlow() {
        return IntegrationFlow.from(inChannel())
                .transform(p -> ">>> send to mqtt, msg is:[" + p + "]")
                .handle(mqttOutbound())
                .get();
    }

    /**
     * 消息消费
     *
     * @return /
     */
    @Bean
    public IntegrationFlow mqttInFlow() {
        return IntegrationFlow.from(mqttInbound())
                .transform(p -> "<<< received from mqtt, msg is:[" + p + "]")
                .handle(logger())
                .get();
    }

    @Bean
    public MessageChannel inChannel() {
        return new DirectChannel();
    }

    @Bean
    public MessageHandler mqttOutbound() {
        MqttPahoMessageHandler messageHandler = new MqttPahoMessageHandler(getClientId() + "_producer", clientFactory);
        messageHandler.setAsync(true);
        messageHandler.setDefaultTopic(properties.getSubscribe());
        return messageHandler;
    }

    @Bean
    public MessageProducerSupport mqttInbound() {
        String[] topics = properties.getTopics().toArray(new String[0]);
        MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(getClientId() + "_consumer", clientFactory, topics);
        adapter.setCompletionTimeout(5000);
        adapter.setConverter(new DefaultPahoMessageConverter());
        adapter.setQos(1);
        return adapter;
    }

    private LoggingHandler logger() {
        LoggingHandler loggingHandler = new LoggingHandler("INFO");
        loggingHandler.setLoggerName("siSample");
        return loggingHandler;
    }

    public String getClientId() {
        String clientId = properties.getClientId();
        if (!StringUtils.hasText(clientId)) {
            clientId = UUID.randomUUID().toString().replaceAll("-", "").toUpperCase(Locale.ENGLISH);
            properties.setClientId(clientId);
        }
        return clientId;
    }

    @Override
    public void sendData(String data) {
        inChannel().send(new GenericMessage<>(data));
    }
}